TypingIndicator simplification #212
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR refactors the TypingIndicator class from an asyncio-based implementation to a threading-based implementation using threading.Timer. The main changes include replacing the continuous asyncio task loop with a single-shot timer and simplifying the start/stop logic.
- Changed from asyncio-based task management to threading-based
Timer - Replaced continuous loop with single timer execution
- Changed default interval from 1 second to 10 units (with unit mismatch in log message)
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| async with self._lock: | ||
| if not self._running: | ||
| return | ||
| logger.debug(f"Starting typing indicator with interval: {self._interval} ms") |
There was a problem hiding this comment.
The log message states the interval is in milliseconds ('ms'), but Python's threading.Timer expects the interval in seconds. The interval value should be documented correctly or converted appropriately.
| logger.debug(f"Starting typing indicator with interval: {self._interval} ms") | |
| logger.debug(f"Starting typing indicator with interval: {self._interval} seconds") |
| func = self._on_timer(context) | ||
| self._timer = Timer(self._interval, func) | ||
| self._timer.start() | ||
| await func() |
There was a problem hiding this comment.
The Timer is created with an async function but Timer expects a synchronous callable. When the timer expires, it will try to call the async function synchronously, which will not execute the coroutine properly. Additionally, the timer only fires once and won't send recurring typing indicators as the original implementation did.
| self._timer.start() | ||
| await func() | ||
|
|
||
| def stop(self) -> None: |
There was a problem hiding this comment.
The stop method signature changed from async def to def, but it's still called with await typing.stop() in agent_application.py line 736. This will cause a runtime error since you cannot await a non-coroutine.
| self._running: bool = False | ||
| self._lock = asyncio.Lock() | ||
| _interval: int | ||
| _timer: Optional[Timer] = None |
There was a problem hiding this comment.
The _timer class attribute will be shared across all instances of TypingIndicator. This should be an instance attribute initialized in __init__ to avoid state being shared between different instances.
| async with self._lock: | ||
| if self._running: | ||
| return | ||
| def __init__(self, interval=10) -> None: |
There was a problem hiding this comment.
The default interval changed from 1 second to 10 seconds without updating call sites. Given the original was 1 second and sent typing indicators continuously in a loop, this significantly changes the behavior and timing characteristics.
This pull request refactors the
TypingIndicatorclass intyping_indicator.pyto replace its asynchronous, asyncio-based implementation with a simpler threading-based approach usingthreading.Timer. The main goal is to simplify the logic for sending "typing" activities by removing the complexity of managing asynchronous tasks and locks.Key changes:
Refactoring from asyncio to threading:
asyncio, including tasks and locks, withthreading.Timerfor scheduling typing indicator events. This removes the need for async task management and simplifies the class. [1] [2]startmethod now usesthreading.Timerto schedule typing events and is no longer protected by an async lock.stopmethod is now synchronous and cancels the timer directly, instead of cancelling an asyncio task.Class interface and logic changes:
_interval._on_timer), which sends the typing activity and handles errors and stopping logic.These changes make the typing indicator logic easier to maintain and less error-prone by avoiding complex async patterns.